package threadpoolnio.server;

import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.concurrent.ExecutorService;

/**
 * @author ：xx
 * @date ：Created in 2020/3/12 20:09
 * @description：
 * @modified By：
 */
public class NIOServer extends Thread {
    final int TIMEOUT = 3000;
    final int PORT = 9999;
    final int BUF_SIZE = 1024;
    Selector selector = null;
    ServerSocketChannel ssc = null;
    private boolean isContine = true;
    private ExecutorService threadPool=null;
    //使用Map保存每个连接，当OP_READ就绪时，根据key找到对应的文件对其进行写入
    Map<SelectionKey, FileChannel> fileMap = new HashMap<SelectionKey, FileChannel>();
    Map<SelectionKey, Long> timeMap=new HashMap<>();
    public static void main(String[] args) {

//        selector();
    }

    @Override
    public void run() {
        System.out.println("=============服务器已启动================");
        selector();
    }

    public void selector() {
        try {
//        第一步：创建Selector
            selector = Selector.open();
//        第二步：创建ServerSocketChannel，绑定监听端口
            ssc = ServerSocketChannel.open();
            ssc.socket().bind(new InetSocketAddress(PORT));
//        第三步：将Channel设置为非阻塞模式
            ssc.configureBlocking(false);
//        第四步：将Channel注册到Selector上，监听连接事件
            ssc.register(selector, SelectionKey.OP_ACCEPT);
//        第五步：循环调用Selector的select方法，检测就绪情况
            while (isContine) {
                if (selector.select(TIMEOUT) == 0) {
//                    System.out.println("===等待连接===");
                    continue;
                }
                //  第六步：调用selectedKeys方法获取就绪channel集合
                Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
                while (iterator.hasNext()) {
                    SelectionKey key = iterator.next();
                    //   第七步：判断就绪事件种类，调用业务处理方法
                    if (key.isValid() && key.isAcceptable()) {
//                        System.out.println("===========连接===========");
                        handleAccept(key);
                    }
                    if (key.isValid() && key.isConnectable()) {
                        System.out.println("isConnectable = true");
                    }
                    if (key.isValid() && key.isReadable()) {
                        handleRead(key);
                    }
                    if (key.isValid() && key.isWritable()) {
//                        System.out.println("===========写===========");
                        handleWrite(key);
                    }
                    iterator.remove();
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public void handleAccept(SelectionKey key) throws IOException {
        ServerSocketChannel ssChannel = (ServerSocketChannel) key.channel();
        SocketChannel sc = ssChannel.accept();
        sc.configureBlocking(false);
        sc.register(key.selector(), SelectionKey.OP_READ, ByteBuffer.allocateDirect(BUF_SIZE));
        System.out.println("===========" + sc.getRemoteAddress() + "连接===========");
    }

    public void handleRead(SelectionKey key) throws IOException {
        //拿到Selector 中channel
        SocketChannel sc = (SocketChannel) key.channel();
        try {
            ByteBuffer buf = (ByteBuffer) key.attachment();
            int bytesRead = -1;
            boolean isFile = false;
            FileChannel fileChannel = fileMap.get(key);
            int currReadTimes = 0;
            while ((bytesRead=sc.read(buf)) > 0) {
                buf.flip();
                //获取buff头部信息，判定数据类型
                int type = buf.getInt();
                //文本数据
                if (type == 0 && fileChannel == null) {
                    isFile=false;
                    System.out.println("===========" + sc.getRemoteAddress() + "文本消息=================");
                }
                //1|| 2 文件数据
                if (type == 1 || type == 2 || fileChannel != null) {
                    isFile = true;
                }
                if (type == 1 && fileChannel == null) {
                    System.out.println("=============" + sc.getRemoteAddress() + "文件数据=============");
                    String filename = null;
                    //巧妙的通过文件头，发送过来文件名字和大小
                    int fileNameLength = buf.getInt();
                    long fileSize = buf.getLong();
                    byte[] nameBytes = new byte[fileNameLength];
                    buf.get(nameBytes);
                    filename = new String(nameBytes, 0, fileNameLength, "utf-8");
                    System.out.println("文件名字：" + filename + ",长度：" + fileSize);
                    //新建文件通道，准备写，
                    if (filename != null && filename.length() > 0) {
                        String currPath = System.getProperty("user.dir");
                        currPath = currPath + File.separator
                                + "upload" + File.separator + filename;
                        File file = new File(currPath);
                        if (!file.getParentFile().exists()) {
                            file.mkdirs();
                        }
                        fileChannel = new FileOutputStream(currPath).getChannel();
                        //存储当前通道和时间（计时）,一次轮询过程可能发送不完数据
                        fileMap.put(key, fileChannel);
                        //存起来当前时间
                        timeMap.put(key,System.currentTimeMillis());
                    }
                }
                if(fileChannel!=null)
                    isFile=true;
                if (!isFile) {
                    //为了处理中文做了特殊处理
                    byte[] bytes = new byte[BUF_SIZE];
                    int i = 0;
                    while (buf.hasRemaining()) {
                        bytes[i] = buf.get();
                        i++;
                    }
                    System.out.println(new String(bytes, "GBK"));
                } else {
                    //必须加循环判定，否则造成文件输出变大问题
                    while (buf.hasRemaining()) {
                        fileChannel.write(buf);
                    }
                }
                buf.clear();
                currReadTimes++;
    //            System.out.println("当前读取次数：" + currReadTimes + ",读取的字节数：" + bytesRead);
            }
//        if(fileChannel!=null)
//            System.out.println("读取结束：读取次数：" + currReadTimes+ ",读取的字节数：" + bytesRead+
//        ",已读取："+fileChannel.position()+"，总大小:"+fileChannel.size());
            if (bytesRead == -1) {
                if(fileChannel!=null)
                    isFile=true;
                if (isFile) {
                    buf.clear();
                    Long startTime=timeMap.get(key);
                    long time=System.currentTimeMillis()-startTime;
                    String result=sc.getRemoteAddress() + "文件上传成功,耗时："+time/1000+"s";
                    System.out.println(result);
                   //回送成功消息给客户端
                    buf.put(result.getBytes("GBK"));
                    buf.flip();
                    sc.write(buf);
                    sc.shutdownOutput();
                }
                if(fileChannel!=null) {
                    fileChannel.close();
                }
                buf.clear();
    //            sc.close();
                key.cancel();
    //            return;
            }
            // Channel的read方法可能返回0，返回0并不一定代表读取完了。
            // 工作线程结束对通道的读取，需要再次更新键的ready集合，将感兴趣的集合重新放在里面
//        key.interestOps(SelectionKey.OP_READ);
//        // 调用wakeup，使得选择器上的第一个还没有返回的选择操作立即返回即重新select
//        key.selector().wakeup();
        } catch (IOException e) {
            System.out.println("=============="+sc.getRemoteAddress()+"已离线===========,原因："+e.getMessage());
            sc.close();
            key.cancel();
        }
    }

    public void handleWrite(SelectionKey key) throws IOException {
        ByteBuffer buf = (ByteBuffer) key.attachment();
        buf.flip();
        SocketChannel sc = (SocketChannel) key.channel();
        while (buf.hasRemaining()) {
            sc.write(buf);
        }
        buf.compact();
    }

    public void close() {
        try {
            isContine = false;
            Thread.sleep(1000);
            if (selector != null) {
                selector.close();
            }
            if (ssc != null) {
                ssc.close();
            }
            System.out.println("===========服务器已停止===================");
        } catch (IOException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}
